package com.demo.utils;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.demo.entity.DataParse;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;

@Component
@Slf4j
public class SseUtils {

    // 构建统一返回的data格式
    public static DataParse dataParse(String type, Object detail) {
//        Map<String, Object> jsonMap = new HashMap<>();
//        jsonMap.put("type", type);
//        jsonMap.put("detail", detail);

        return new DataParse(type, detail);
    }
    

    // Sse发送消息
    public static void sendSseMessage(SseEmitter emitter, String id, String event, DataParse data, Long sleepTime) {
        SseEmitter.SseEventBuilder eventBuilder = SseEmitter.event();

        if (ObjectUtils.isNotEmpty(id)) {
            eventBuilder.id(id);
        }

        if (ObjectUtils.isNotEmpty(event)) {
            eventBuilder.name(event);
        }

        if (ObjectUtils.isNotEmpty(data)) {
            eventBuilder.data(data);
        }

        try {
            emitter.send(eventBuilder);

            Thread.sleep(sleepTime);
        } catch (IOException e) {
            handleError(emitter, e);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            SseUtils.handleError(emitter, e);
        }
    }

    public static void sendSseMessage(SseEmitter emitter, String id, String event, String data, Long sleepTime) {
        SseEmitter.SseEventBuilder eventBuilder = SseEmitter.event();

        if (ObjectUtils.isNotEmpty(id)) {
            eventBuilder.id(id);
        }

        if (ObjectUtils.isNotEmpty(event)) {
            eventBuilder.name(event);
        }

        if (ObjectUtils.isNotEmpty(data)) {
            eventBuilder.data(data, MediaType.TEXT_PLAIN);
        } else {
            eventBuilder.data("", MediaType.TEXT_PLAIN);
        }
        try {
            emitter.send(eventBuilder);
            Thread.sleep(sleepTime);
        } catch (IllegalStateException e) {
            // 如果emitter已经完成，记录调试日志并抛出，让调用者处理
            if (e.getMessage().contains("ResponseBodyEmitter has already completed")) {
                log.debug("SSE emitter已经关闭，忽略消息发送: {}", event, e);
                throw e; // 重新抛出异常，由调用者处理
            } else {
                log.error("发送SSE消息失败: {}", event, e);
                handleError(emitter, e);
            }
        } catch (IOException e) {
            log.error("发送SSE消息IO异常", e);
            handleError(emitter, e);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("发送SSE消息被中断", e);
            SseUtils.handleError(emitter, e);
        }
    }

    public static void sendSseMessage(SseEmitter emitter, String id, String event, Object data, Long sleepTime) throws JsonProcessingException {
        SseEmitter.SseEventBuilder eventBuilder = SseEmitter.event();

        if (ObjectUtils.isNotEmpty(id)) {
            eventBuilder.id(id);
        }

        if (ObjectUtils.isNotEmpty(event)) {
            eventBuilder.name(event);
        }

        if (ObjectUtils.isNotEmpty(data)) {
            ObjectMapper mapper = new ObjectMapper();
            String json = mapper.writeValueAsString(data); // 数组转为JSON字符串
            eventBuilder.data(json, MediaType.TEXT_PLAIN);
        } else {
            eventBuilder.data("", MediaType.TEXT_PLAIN);
        }
        try {
            emitter.send(eventBuilder);
            Thread.sleep(sleepTime);
        } catch (IllegalStateException e) {
            // 如果emitter已经完成，记录调试日志并抛出，让调用者处理
            if (e.getMessage().contains("ResponseBodyEmitter has already completed")) {
                log.debug("SSE emitter已经关闭，忽略消息发送: {}", event, e);
                throw e; // 重新抛出异常，由调用者处理
            } else {
                log.error("发送SSE消息失败: {}", event, e);
                handleError(emitter, e);
            }
        } catch (IOException e) {
            log.error("发送SSE消息IO异常", e);
            handleError(emitter, e);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("发送SSE消息被中断", e);
            SseUtils.handleError(emitter, e);
        }
    }

    public static void sendSseMessage(SseEmitter emitter, ServerSentEvent<Object> serverSentEvent, Long sleepTime){
        SseEmitter.SseEventBuilder eventBuilder = SseEmitter.event();

        if (ObjectUtils.isNotEmpty(serverSentEvent.id())) {
            eventBuilder.id(serverSentEvent.id());
        }

        if (ObjectUtils.isNotEmpty(serverSentEvent.event())) {
            eventBuilder.name(serverSentEvent.event());
        }

        if (ObjectUtils.isNotEmpty(serverSentEvent.data())) {
            String json = JSON.toJSONString(serverSentEvent.data());
            eventBuilder.data(json, MediaType.TEXT_PLAIN);
        } else {
            eventBuilder.data("", MediaType.TEXT_PLAIN);
        }
        try {
            emitter.send(eventBuilder);
            Thread.sleep(sleepTime);
        } catch (IllegalStateException e) {
            // 如果emitter已经完成，记录调试日志并抛出，让调用者处理
            if (e.getMessage().contains("ResponseBodyEmitter has already completed")) {
                log.debug("SSE emitter已经关闭，忽略消息发送: {}", serverSentEvent.event(), e);
            } else {
                log.error("发送SSE消息失败: {}", serverSentEvent.event(), e);
                handleError(emitter, e);
            }
        } catch (IOException e) {
            log.error("发送SSE消息IO异常", e);
            handleError(emitter, e);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("发送SSE消息被中断", e);
            SseUtils.handleError(emitter, e);
        }
    }

    public static void sendError(SseEmitter emitter, String id, String event, DataParse data, Throwable error) {
        sendSseMessage(emitter, id, event, data, 0L);
        emitter.completeWithError(error);
    }

    public static void sendError(SseEmitter emitter, String id, String event, String data, Throwable error) {
        sendSseMessage(emitter, id, event, data, 0L);
        emitter.completeWithError(error);
    }

    // 错误处理方法
    public static void handleError(SseEmitter emitter, Throwable error) {
        try {
            emitter.send(SseEmitter.event().name("error").data(error.getMessage()));
            emitter.completeWithError(error);
        } catch (IOException e) {
            emitter.completeWithError(e);
        }
    }

    // 完成处理方法
    public static void completeEmitter(SseEmitter emitter, String id) {
        // 完成发送finish
        sendSseMessage(emitter, id, "finish", new DataParse("finish", null), 0L);
        try {
            emitter.complete();
        } catch (Exception e) {
            // 静默处理
        }
    }

    public static void completeEmitterNullData(SseEmitter emitter, String requestId) {
        // 完成发送finish
        try {
            // 使用明确的UTF-8编码
            emitter.send(SseEmitter.event()
                    .id(requestId)
                    .name("finish")
                    .data("", MediaType.parseMediaType("text/event-stream;charset=UTF-8")));

            emitter.complete();
        } catch (IllegalStateException e) {
            // 如果是由于ResponseBodyEmitter已经完成导致的异常，则静默处理
            if (e.getMessage().contains("ResponseBodyEmitter has already completed")) {
                log.debug("SSE emitter已经关闭，忽略finish事件发送", e);
            } else {
                log.error("发送finish事件失败", e);
            }
            // 确保emitter完成
            try {
                emitter.complete();
            } catch (Exception ignored) {
                // 静默处理
            }
        } catch (IOException e) {
            // 处理IO异常
            log.error("发送finish事件失败 - IO异常", e);
            // 确保emitter完成
            try {
                emitter.complete();
            } catch (Exception ignored) {
                // 静默处理
            }
        } catch (Exception e) {
            // 静默处理其他异常
            log.error("发送finish事件失败", e);
            // 确保emitter完成
            try {
                emitter.complete();
            } catch (Exception ignored) {
                // 静默处理
            }
        }
    }

    // Emitter回调配置
    public static void setupEmitterCallbacks(SseEmitter emitter) {
        emitter.onTimeout(() -> log.warn("客户端响应超时"));

        emitter.onCompletion(() -> log.info("SSE连接正常关闭"));

        emitter.onError(error -> log.error("SSE连接异常终止", error));
    }

    /**
     * sse 结果数据转成json数据
     *
     * @param sse
     * @return
     */
    public static JSONObject sseToJson(String sse) {
        // 保持插入顺序
        JSONObject result = new JSONObject(true);
        String[] lines = sse.split("\n");
        for (String line : lines) {
            String[] parts = line.split(":", 2);
            if (parts.length == 2) {
                String key = parts[0].trim();
                String value = parts[1].trim();

                if ("data".equals(key)) {
                    try {
                        JSONObject nested = JSON.parseObject(value);
                        result.put(key, nested);
                    } catch (Exception e) {
                        result.put(key, value);
                    }
                } else {
                    result.put(key, value);
                }
            }
        }
        log.info("sseToJson result --" + JSON.toJSONString(result, true));
        return result;
    }

    /**
     * sse 结果数据转成json数据
     * @param sse
     * @return
     */
    public static JSONObject streamToJson(String sse){
        // 保持插入顺序
        JSONObject result = new JSONObject(true);
        String[] lines = sse.split("\n");
        for (String line : lines) {
            String[] parts = line.split(":", 2);
            if (parts.length == 2) {
                String key = parts[0].trim();
                String value = parts[1].trim();
                result.put(key, value);
            }
        }
        result.putIfAbsent("data", "");
        log.info("streamToJson result --" + JSON.toJSONString(result, true));
        return result;
    }


    /**
     * 从 sse 结果中获取data数据
     * @param sse
     * @return
     */
    public static String getStreamData(String sse) {
        String data = "";
        String[] lines = sse.split("\n");
        for (String line : lines) {
            if (line.contains("data:")) {
                data = line.replace("data:", "");
                break;
            }
        }
//        log.info("getStreamData data:" + data);
        return data;
    }
}
